package org.ala.distributed_transaction_commons.communication;

import org.ala.distributed_transaction_commons.communication.coordinator.CoordinatorActionExector;
import org.ala.distributed_transaction_commons.communication.coordinator.CoordinatorSubscriber;
import org.ala.distributed_transaction_commons.communication.participant.ParticipantPublisher;
import org.ala.distributed_transaction_commons.logs.LoggerFactory;
import org.ala.distributed_transaction_commons.redis.DistributedTransactionRedisTools;
import org.slf4j.Logger;
import org.springframework.beans.factory.InitializingBean;


/**
 * 协调者
 *
 * @author 骆毅(ala)
 * @date 2024年2月18日
 */
public class Coordinator implements InitializingBean {
	
	
	static Logger log = LoggerFactory.getLogger("distributed_transaction/coordinator", "coordinator");

	
	/**
	 * redis工具
	 */
	protected DistributedTransactionRedisTools distributedTransactionRedisTools;
	/**
	 * 发送给参与者的publisher
	 */
	protected ParticipantPublisher publisher;
	/**
	 * 发送给协调者的subscriber
	 */
	protected CoordinatorSubscriber subscriber;
	/**
	 * message对应的action执行器
	 */
	protected CoordinatorActionExector exector;
	

	@Override
	public void afterPropertiesSet() throws Exception {
		//	初始化publisher
		initPublisher();
		//	初始化exector
		initExector();
		//	初始化subscrib
		initSubscrib();
	}
	
	/**
	 * 初始化exector
	 * @throws Exception 
	 */
	protected void initExector() throws Exception {
		exector = new CoordinatorActionExector();
		exector.setDistributedTransactionRedisTools(distributedTransactionRedisTools);
		exector.setParticipantPublisher(publisher);
		exector.afterPropertiesSet();
	}
	/**
	 * 初始化publisher
	 * @throws Exception 
	 */
	protected void initPublisher() throws Exception {
		publisher = new ParticipantPublisher();
		publisher.setDistributedTransactionRedisTools(distributedTransactionRedisTools);
		publisher.afterPropertiesSet();
	}
	
	
	/**
	 * 初始化subscrib
	 * @throws Exception 
	 */
	protected void initSubscrib() throws Exception {
		subscriber = new CoordinatorSubscriber();
		subscriber.setDistributedTransactionRedisTools(distributedTransactionRedisTools);
		subscriber.setOnSubscrib(channel -> {
			log.info("[Coordinator setOnSubscrib] 协调者监听器，开始监听channel:" + channel);
		});
		subscriber.setOnUnSubscrib(channel -> {
			log.info("[Coordinator setOnSubscrib] 协调者监听器，取消监听channel:" + channel + " 重启监听");
			subscriber.reSubscrib(new String[] {channel});
		});
		subscriber.setOnMessage(msg -> {
			log.debug("[Coordinator setOnMessage] 协调者监听器，收到消息 to:" + msg.getTo() + " biz:" + msg.getBiz() + " type:" + msg.getMsgType() + " txId:" + msg.getTxId() + " pid:" + msg.getPid());
			exector.doAction(msg);
		});
		subscriber.afterPropertiesSet();
		//	开启监听
		String[] channels = Message.allChannelCoordinator();
		subscriber.reSubscrib(channels);
	}
	

	public DistributedTransactionRedisTools getDistributedTransactionRedisTools() {
		return distributedTransactionRedisTools;
	}

	public void setDistributedTransactionRedisTools(DistributedTransactionRedisTools distributedTransactionRedisTools) {
		this.distributedTransactionRedisTools = distributedTransactionRedisTools;
	}

	public CoordinatorActionExector getExector() {
		return exector;
	}

	public void setExector(CoordinatorActionExector exector) {
		this.exector = exector;
	}

	public ParticipantPublisher getPublisher() {
		return publisher;
	}

	public void setPublisher(ParticipantPublisher publisher) {
		this.publisher = publisher;
	}

}
